package org.argus.job.worker.core;

import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;

import lombok.extern.slf4j.Slf4j;
import org.argus.job.common.model.WorkerAppInfo;
import org.argus.job.common.utils.CommonUtils;
import org.argus.job.common.utils.HttpUtils;
import org.argus.job.remote.framework.base.Address;
import org.argus.job.remote.http.HttpVertxCSInitializer;
import org.argus.job.worker.core.background.OmsLogHandler;
import org.argus.job.worker.core.background.WorkerHealthReporter;
import org.argus.job.worker.core.background.discovery.PowerJobServerDiscoveryService;
import org.argus.job.worker.core.background.discovery.ServerDiscoveryService;
import org.argus.job.worker.core.common.PowerJobWorkerConfig;
import org.argus.job.worker.core.common.WorkerRuntime;
import org.argus.job.worker.core.executor.ExecutorManager;
import org.argus.job.worker.core.extension.processor.ProcessorFactory;
import org.argus.job.common.network.IServiceInstanceIPParser;
import org.argus.job.worker.core.persistence.DbTaskPersistenceService;
import org.argus.job.worker.core.persistence.TaskPersistenceService;
import org.argus.job.worker.core.processor.impl.BuiltInDefaultProcessorFactory;
import org.argus.job.worker.core.processor.loader.PowerJobProcessorLoader;
import org.argus.job.worker.core.processor.loader.ProcessorLoader;

/**
 * 客户端启动类
 *
 * @author KFCFans
 * @since 2020/3/16
 */
@Slf4j
public class ArgusJobWorker {
    protected final WorkerRuntime workerRuntime;

    private final IServiceInstanceIPParser ipParser;
    private final AtomicBoolean initialized = new AtomicBoolean(false);

    public ArgusJobWorker(PowerJobWorkerConfig config, IServiceInstanceIPParser ipParser) {
        this.workerRuntime = new WorkerRuntime();
        workerRuntime.setWorkerConfig(config);
        this.ipParser = ipParser;
    }

    public void init() throws Exception {

        if (!initialized.compareAndSet(false, true)) {
            log.warn("[PowerJobWorker] please do not repeat the initialization");
            return;
        }

        Stopwatch stopwatch = Stopwatch.createStarted();
        log.info("[PowerJobWorker] start to initialize PowerJobWorker...");

        PowerJobWorkerConfig config = workerRuntime.getWorkerConfig();
        CommonUtils.requireNonNull(config, "can't find PowerJobWorkerConfig, please set PowerJobWorkerConfig first");

        ServerDiscoveryService serverDiscoveryService = new PowerJobServerDiscoveryService(config);
        workerRuntime.setServerDiscoveryService(serverDiscoveryService);

        try {

            // 校验 appName
            HttpUtils.setAppName(config.getAppName());
            WorkerAppInfo appInfo = serverDiscoveryService.assertApp();
            workerRuntime.setAppInfo(appInfo);

            // 在发第一个请求之前，完成真正 IP 的解析
            int externalPort = config.getPort();

            // 初始化网络数据，区别对待上报地址和本机绑定地址（对外统一使用上报地址）
            String externalIp = ipParser.getInstanceIp();
            log.info("[PowerJobWorker] [ADDRESS_INFO] externalIp: {}, externalPort: {}", externalIp, externalPort);
            workerRuntime.setWorkerAddress(Address.toFullAddress(externalIp, Integer.valueOf(externalPort)));

            // 初始化 线程池
            final ExecutorManager executorManager = new ExecutorManager(workerRuntime.getWorkerConfig());
            workerRuntime.setExecutorManager(executorManager);

            // 初始化 ProcessorLoader
            ProcessorLoader processorLoader = buildProcessorLoader(workerRuntime);
            workerRuntime.setProcessorLoader(processorLoader);

            // TODO transporter 初始化成统一的 httpclient？
            HttpVertxCSInitializer httpVertxCSInitializer = new HttpVertxCSInitializer();
            httpVertxCSInitializer.init(null);
            workerRuntime.setTransporter(httpVertxCSInitializer.buildTransporter());

            // 连接 server，在Server网关侧做高可用，不需要和具体Job Server实例中的一个建立连接并保活，所以这里不需要做定时Server可达性检查
            // serverDiscoveryService.timingCheck(workerRuntime.getExecutorManager().getCoreExecutor());

            log.info("[PowerJobWorker] PowerJobRemoteEngine initialized successfully.");

            // 初始化日志系统
            OmsLogHandler omsLogHandler =
                new OmsLogHandler(workerRuntime.getWorkerAddress(), workerRuntime.getTransporter(), serverDiscoveryService);
            workerRuntime.setOmsLogHandler(omsLogHandler);

            // 初始化存储
            TaskPersistenceService taskPersistenceService =
                new DbTaskPersistenceService(workerRuntime.getWorkerConfig().getStoreStrategy());
            taskPersistenceService.init();
            workerRuntime.setTaskPersistenceService(taskPersistenceService);
            log.info("[PowerJobWorker] local storage initialized successfully.");

            // 初始化定时任务
            workerRuntime.getExecutorManager().getCoreExecutor().scheduleAtFixedRate(
                new WorkerHealthReporter(workerRuntime), 0, config.getHealthReportInterval(), TimeUnit.SECONDS);
            workerRuntime.getExecutorManager().getCoreExecutor().scheduleWithFixedDelay(omsLogHandler.logSubmitter, 0,
                5, TimeUnit.SECONDS);

            log.info("[PowerJobWorker] PowerJobWorker initialized successfully, using time: {}, congratulations!",
                stopwatch);
        } catch (Exception e) {
            log.error("[PowerJobWorker] initialize PowerJobWorker failed, using {}.", stopwatch, e);
            throw e;
        }
    }

    private ProcessorLoader buildProcessorLoader(WorkerRuntime runtime) {
        List<ProcessorFactory> customPF =
            Optional.ofNullable(runtime.getWorkerConfig().getProcessorFactoryList()).orElse(Collections.emptyList());
        List<ProcessorFactory> finalPF = Lists.newArrayList(customPF);

        // 后置添加2个系统 ProcessorLoader
        finalPF.add(new BuiltInDefaultProcessorFactory());
        // container factory 暂时丢失，无状态多节点场景下，需要每次请求下发jar，后续再思考怎么做
        // finalPF.add(new JarContainerProcessorFactory(runtime));

        return new PowerJobProcessorLoader(finalPF);
    }

    public WorkerRuntime getWorkerRuntime() {
        return workerRuntime;
    }

    public void destroy() throws Exception {
        workerRuntime.getExecutorManager().shutdown();
    }
}
